package com.z.redis;

import com.z.redis.datatype.BytesWrapper;
import com.z.redis.datatype.RedisData;
import io.netty.channel.Channel;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author zxb
 */
public class RedisCoreImpl implements RedisCore{
    /**
     * 使用ConcurrentSkipListMap实现,相比较ConcurrentHashMap，支持范围查询，天然有序，适合Redis的SortedSet等有序结构
     * 使用线程安全的NavigableMap实现，适合高并发的场景
     * 客户端可能使用Hash路由，更换成跳表可以更好的避免Hash冲突
     */
    //数据存储核心
//    private final ConcurrentNavigableMap<ByteWrapper, RedisData> map = new ConcurrentHashMap<ByteWrapper, RedisData>();
    private final ConcurrentNavigableMap<BytesWrapper, RedisData> map = new ConcurrentSkipListMap<BytesWrapper, RedisData>();

    //客户端管理 维护客户端名称与连接的关联 支持快速双向查找
    private final ConcurrentHashMap<BytesWrapper, Channel> clients = new ConcurrentHashMap<>();
    private final Map<Channel, BytesWrapper> clientNames = new ConcurrentHashMap<>();

    //键集合获取
    @Override
    public Set<BytesWrapper> keys() { return map.keySet(); }

    // 客户端管理
    /**
     * 维护双向映射关系，支持名称或者Channel快速查找
     */
    @Override
    public void putClient(BytesWrapper connectionName, Channel channelContext){
        clients.put(connectionName, channelContext);
        clientNames.put(channelContext, connectionName);
    }

    //数据是否存在
    @Override
    public boolean exist(BytesWrapper key) { return map.containsKey(key); }
    //数据存取
    @Override
    public void put(BytesWrapper key, RedisData value) { map.put(key, value); }
    @Override
    public RedisData get(BytesWrapper key){
        // 使用 replace 实现原子化的过期检查
        return map.compute(key, (k, v) -> {
            if (v == null || (v.getTimeout() != -1 && v.getTimeout() < System.currentTimeMillis())) {
                return null;
            }
            return v;
        });
    }

    @Override
    public long remove(List<BytesWrapper> keys) {
//        return keys.stream().peek(map::remove).count();
        return keys.stream().filter(k -> map.remove(k) != null).count();
    }
    @Override
    public void cleanAll() { map.clear(); }
}
